package com.shujia.HDFS;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.*;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;

import java.io.*;

public class Demo02HDFSAPI {
    FileSystem fs;

    @Before
    public void init() throws IOException {
        // 创建一个配置文件
        Configuration conf = new Configuration();
        // 设置NameNode的地址
        conf.set("fs.defaultFS", "hdfs://master:9000");
        // 设置副本数量
        conf.set("dfs.replication", "1");

        // 与NameNode建立连接
        fs = FileSystem.get(conf);
    }

    @Test
    // 判断路径存不存在
    public void exists() throws IOException {
        Path path = new Path("/data");
        System.out.println(fs.exists(path));
    }

    @Test
    // 创建一个/TestAPI/test1目录
    // 默认相当于 mkdir -p
    public void mkdirs() throws IOException {
        fs.mkdirs(new Path("/TestAPI/test1"));
        fs.mkdirs(new Path("/TestAPI/test2"));
    }

    @Test
    // 删除目录
    public void delete() throws IOException {
        fs.delete(new Path("/TestAPI/test2"), true);
    }

    @Test
    // 上传文件至/TestAPI/test1
    public void copyFromLocalFile() throws IOException {
        fs.copyFromLocalFile(new Path("data/students.txt"), new Path("/TestAPI/test1"));
    }

    @Test
    // 上传文件至/TestAPI/test1 并移除本地文件
    public void moveFromLocalFile() throws IOException {
        fs.moveFromLocalFile(new Path("data/students.txt"), new Path("/TestAPI/test1/students_new.txt"));
    }

    @Test
    // 将HDFS上的文件下载到本地，并移除HDFS的文件
    public void moveToLocalFile() throws IOException {
        fs.moveToLocalFile(new Path("/TestAPI/test1/students.txt"), new Path("data/"));
    }

    @Test
    // 查看/TestAPI/test1目录下的文件
    public void listStatus() throws IOException {
        FileStatus[] fileStatuses = fs.listStatus(new Path("/TestAPI/test1"));
        for (FileStatus fileStatus : fileStatuses) {
            System.out.println(fileStatus.getPath());
        }
    }


    @Test
    // 读取HDFS上的文件
    public void readFromHDFSFile() throws IOException {
        FSDataInputStream fsDataInputStream = fs.open(new Path("/TestAPI/test1/students_new.txt"));
        BufferedReader br = new BufferedReader(new InputStreamReader(fsDataInputStream));

        String line;
        while ((line = br.readLine()) != null) {
            System.out.println(line);
        }

        br.close();
        fsDataInputStream.close();

    }

    @Test
    // 将数据写入HDFS
    public void writeToHDFSFile() throws IOException {
        FSDataOutputStream fsDataOutputStream;
        // 如果文件存在，则追加，不存在，则创建
        Path path = new Path("/TestAPI/test1/newFile.txt");
        if (fs.exists(path)) {
            System.out.println("文件已存在，进行追加");
            fsDataOutputStream = fs.append(path);
        } else {
            System.out.println("创建文件");
            fsDataOutputStream = fs.create(path);
        }

        // 写入文件
        BufferedWriter bw = new BufferedWriter(new OutputStreamWriter(fsDataOutputStream));

        bw.write("1500100982,钱若芹,21,女,理科二班");
        bw.newLine();
        bw.write("1500100983,左傲薇,22,女,理科四班");
        bw.newLine();
        bw.write("1500100973,钮幼南,23,女,文科一班");
        bw.newLine();

        bw.flush();

        bw.close();
        fsDataOutputStream.close();

    }


    @After
    public void close() throws IOException {
        fs.close();
    }
}
